package com.atguigu.networkflow_analysis


import java.lang

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.RichAllWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

import scala.util.hashing.MurmurHash3
import com.google.common.base.Optional
import com.google.common.hash.HashCode


/**
 * UV统计(使用布隆过滤器存在redis中)
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.networkflow_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/12 11:12
 */
object UvWithBloomFilter {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val path: String = UvWithBloomFilter.getClass.getClassLoader.getResource("UserBehavior.csv").getPath

    val value: DataStream[UvCount] = env.readTextFile(path)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          UserBehavior(dataArrays(0).trim.toLong, dataArrays(1).trim.toLong, dataArrays(2).trim.toInt, dataArrays(3).trim, dataArrays(4).trim.toLong)
        }
      )
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[UserBehavior](Time.seconds(5)) {
        override def extractTimestamp(element: UserBehavior): Long = element.timestamp * 1000L
      })
      .filter(_.behavior == "pv")
      .timeWindowAll(Time.hours(1))
//      .trigger(new MyTrigger()) //设置触发器
      .apply(new MyBloomWindow())

    value.print("output").setParallelism(1)

    env.execute("UvWithBloomFilter job")
  }
}

class MyTrigger() extends Trigger[UserBehavior, TimeWindow]{
  //来一条数据触发一次window窗口的计算并关闭
  //调用添加到窗格中的每个元素。这将决定结果是否评估窗格以发出结果
  override def onElement(element: UserBehavior, timestamp: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.FIRE_AND_PURGE

  //使用触发器上下文设置的处理时间计时器触发时调用。
  override def onProcessingTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  //当使用触发器上下文设置的事件时间计时器触发时调用。
  override def onEventTime(time: Long, window: TimeWindow, ctx: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

  //清除触发器对给定窗口可能仍然保持的任何状态。
  override def clear(window: TimeWindow, ctx: Trigger.TriggerContext): Unit = {}
}


class MyBloomWindow() extends RichAllWindowFunction[UserBehavior, UvCount, TimeWindow]{

  var jedis: Jedis = _

  override def open(parameters: Configuration): Unit = {
    jedis = new Jedis("hadoop105", 6379)
  }

  override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {

    //一个窗口一个bloom位图
    val storeKey = window.getEnd.toString

    //计算总访问量
    var count: Long = 0L

    val str = jedis.hget("count", storeKey)
    if (str != null) {
      count = str.toLong
    }


    for (elem <- input) {
      //计算该id在当前窗口中的位图位置
      val id: String = elem.userId.toString
      val bloom = new Bloom()
      val hash: Long = bloom.hashWithMurHash(id)

      //获取对应hash值的位图值
      val boolean: lang.Boolean = jedis.getbit(storeKey, hash)
      //如果位置为0，表名不存在该用户
      if (!boolean){
        jedis.setbit(storeKey, hash, true)

        count = count + 1
      }
    }

    //设置过期时间
    jedis.expire(storeKey, 100)

    //把结果保存进redis的count中
    jedis.hset("count", storeKey, count.toString)

    out.collect(UvCount(window.getEnd, count))
  }

  override def close(): Unit = {
    if (jedis != null) {
      jedis.close()
    }
  }
}


//布隆位图过滤器(默认处理亿级数据)
class Bloom(size: Long = 0){

  //位图的总大小,默认75M  2^27位
  private val cap = if (size > 0) size else 1 << 27

  /**
   * 使用SMHasher套件求取字符串的hash值
   * @param str
   * @return
   */
  def hashWithMurHash(str: String): Long ={
    //计算字符串的高质量散列
    val result: Int = MurmurHash3.stringHash(str)

    result & (cap - 1)
  }

  /**
   * 使用自定义hash算法
   * (有报错，String必须要是偶数位)
   *
   * @param str 需要计算的hash字符串
   * @param seed  计算种子
   * @return
   */
  def hash(str: String, seed: Int): Long ={
    //保存结果
    var result: Long = 0L
    //对一个字符串进行hash运算
    for (i <- 0 until str.length) {
      result = result * seed + str.charAt(i)
    }
    //进行与运算，把数据平均分在0-26位之间
    result & (cap - 1)
  }

  /**
   * 使用Google的guava计算hashcode
   * @param str
   * @return
   */
  def hashWithGoogleGashcode(str: String): Long ={
    val result: Int = HashCode.fromString(str).hashCode()

    result & (cap - 1)
  }

}
